package com.study.flink.sink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}


/**
  * Redis Sink Demo
  * 在redis中查看存储的数据: HGET KEY_NAME FIELD
  * 示例：HEGT word_count cc
  */
object ScalaRedisSinkDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取数据
    val text = env.socketTextStream("localhost", 9999)

    // 3 transformation
    import org.apache.flink.api.scala._
    val wordAndCountStream = text.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)

    // 4 输出到Redis
    val conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.8.37").setPort(6379).setPassword("redispwd").build()
    wordAndCountStream.addSink(new RedisSink(conf, new TestRedisMapper()))

    // 5 启动
    env.execute("Redis Sink Demo")

  }
}

class TestRedisMapper extends RedisMapper[(String, Int)] {

  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET, "word_count")
  }

  override def getKeyFromData(wordAndCount: (String, Int)): String = {
    wordAndCount._1
  }

  override def getValueFromData(wordAndCount: (String, Int)): String = {
    wordAndCount._2.toString
  }
}
